
--[[
*********************************************************************************
** 
** Copyright 2021-2024 LuaOS
**
** THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
** IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
** FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
** AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
** LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
** OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
** SOFTWARE.
** 
*********************************************************************************
]]--

--------------------------------------------------------------------------------

local format = string.format;
local proto_type = require("luaos.cluster.protocol");
local header = require("luaos.cluster.header");
local active_sessions = {};

local identifiers  = {};
local free_numbers = std.list();

--------------------------------------------------------------------------------

local function check_expires(timer, expires)
  local now = os.system_clock("ms");
  local removed = {}
  for k, v in pairs(identifiers) do
    if not v.online then
	  if now - v.offline_time > 86400 then
	    table.insert(removed, k);
	  end
	end
  end
  for _, v in ipairs(removed) do
    free_numbers:push_back(identifiers[v].id);
    identifiers[v] = nil;
  end
  return expires;
end

--------------------------------------------------------------------------------

local function save_identifiers()
  local data = wrap(identifiers);
  local file = io.open("cluster.master.db", "wb");
  if file then
    file:write(data);
	file:close();
  end
end

--------------------------------------------------------------------------------

local function load_identifiers()
  local file = io.open("cluster.master.db", "rb");
  if file then
    local data = file:read("a");
	file:close();
    identifiers = unwrap(data);
  end
end

--------------------------------------------------------------------------------

local function init_identifiers()
  load_identifiers();
  for k, v in pairs(identifiers) do
    v.online = false;
	v.offline_time = os.system_clock("ms");
  end
  for i = 1, 65535 do
    local found = false;
    for k, v in pairs(identifiers) do
      if i == v.id then
  	  found = true;
  	  break;
  	end
    end
    if not found then
      free_numbers:push_back(i);
    end
  end
  local timer = os.timer();
  timer:expires(1000, bind(check_expires, timer, 1000));
end

--------------------------------------------------------------------------------

local function encode(t)
  return proto_type.encode(t);
end

--------------------------------------------------------------------------------

local function make_peer_id(session)
  local identifier = session.identifier;
  local data = identifiers[identifier];
  if data then
	data.online = true;
	data.offline_time = nil;
    return data.id, false;
  end
  local id = free_numbers:front();
  free_numbers:pop_front();
  identifiers[identifier] = { id = id, online = true };
  return id, true;
end

--------------------------------------------------------------------------------

local function get_peer_id(session)
  local identifier = session.identifier;
  local data = identifiers[identifier];
  return data and data.id or nil;
end

--------------------------------------------------------------------------------

local function send_members(session)
  local peerid, newnode = make_peer_id(session);
  print(format("cluster node number: %d", peerid));
  if not newnode then
    return;
  end
  local packet = {
	what   = proto_type.number,
	number = peerid;
  };
  local socket = session.socket;
  socket:send(encode(packet));  
  for peer, member in pairs(active_sessions) do
    if peer:id() ~= socket:id() then
      local packet = {
        what = proto_type.forword,
        id   = get_peer_id(member);
        ip   = member.ip,
        port = member.port,
      };
      socket:send(encode(packet));
    end
  end	
  socket:send(encode({what = proto_type.ready}));
end

--------------------------------------------------------------------------------

local function on_error(peer, msg)
  local session = active_sessions[peer];
  if session then
    local identifier = session.identifier;
    local data = identifiers[identifier];
    if data then
	  data.online = false;
	  data.offline_time = os.system_clock("ms");
	end
    peer:close();
    active_sessions[peer] = nil;
  end
end

--------------------------------------------------------------------------------

local function on_receive(peer, data, ec)
  if ec then
    on_error(peer, ec);
    return;
  end
end

--------------------------------------------------------------------------------

local function new_session(peer)
  local port = peer:getheader(header.port);
  if not port then
    return false;
  end
  local address = peer:getheader(header.address);
  if not address then
    address = peer:endpoint().address;
  end
  local identifier = peer:getheader(header.identifier);
  if not identifier then
	return false;
  end
  for k, v in pairs(active_sessions) do
    if v.identifier == identifier then
	  error(format("identifier already exists: %s", identifier));
	  return false;
	end
  end
  local session = {
    socket     = peer,
    ip         = address,
    port       = port,
	identifier = identifier
  };
  send_members(session);
  active_sessions[peer] = session;
  save_identifiers();
  print(format("cluster member accept from %s", address));
  return true;
end

--------------------------------------------------------------------------------

local function on_accept(peer, ec)
  if ec then
    on_error(peer, ec);
    return;
  end
  if not new_session(peer) then
    peer:close();
    return;
  end
  peer:receive(bind(on_receive, peer));
end

--------------------------------------------------------------------------------

local function main(port, host)
  port = port or 80;
  local server = io.server("ws");
  local ok = server:listen(host or "0.0.0.0", port, on_accept);  
  if not ok then
    error(format("server listen error at port: %d", port));
    return;
  end
  init_identifiers();
  print(format("%s works on port %d", os.name(), port));  
  while not os.stopped() do
    os.wait();
  end
  save_identifiers();
  server:close();
end

--------------------------------------------------------------------------------

luaos.start(main, ...);

--------------------------------------------------------------------------------
